package com.cdy.server.handler;

import com.cdy.common.command.*;
import com.cdy.store.manager.FileSystem;
import com.cdy.store.repository.FileRepository;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 服务端命令处理器
 * <p>
 * Created by 陈东一
 * 2019/10/20 0020 12:30
 */
@Slf4j
public class HandlerManager implements CommandHandler {

    public final HashMap<String, CommandHandler> handlerMap = new HashMap<>();
    private final FileRepository fileRepository;
    private final FileSystem fileSystem;

    private int highReadWaterMarker = 900, lowReadWaterMarker = 600;
    private ThreadPoolExecutor executor =
            new ThreadPoolExecutor(8, 16, 1, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(1000), new ThreadPoolExecutor.DiscardOldestPolicy());

//    private ScheduledThreadPoolExecutor schedule = new ScheduledThreadPoolExecutor(1);

    public HandlerManager(FileRepository fileRepository, FileSystem fileSystem) {
        this.fileRepository = fileRepository;
        this.fileSystem = fileSystem;
        handlerMap.put(RequestUploadCommand.class.getName(), new RequestUploadCommandHandler(fileRepository));
        handlerMap.put(UploadCommand.class.getName(), new UploadCommandHandler(fileRepository, fileSystem));
        handlerMap.put(RequestDownloadCommand.class.getName(), new RequestDownloadCommandHandler(fileRepository));
        handlerMap.put(DownloadCommand.class.getName(), new DownloadCommandHandler(fileRepository, fileSystem));

    }

    @Override
    public void handle(ChannelHandlerContext ctx, Command msg) {

            int queued = executor.getQueue().size();
            if (queued > highReadWaterMarker) {
                ctx.channel().config().setAutoRead(false);
            }
            if (queued < lowReadWaterMarker) {
                ctx.channel().config().setAutoRead(true);
            }

        executor.execute(()-> {
            CommandHandler commandHandler = handlerMap.get(msg.getClass().getName());
            if (commandHandler != null) {
                commandHandler.handle(ctx, msg);
            } else {
                log.warn("no such CommandHandler {} ", msg);
            }
        });
    }
}
